package com.hsylient.canal.listener;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import jakarta.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class CanalListener {

    @Autowired
    private CanalConnector canalConnector;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @PostConstruct
    public void listen() {
        new Thread(() -> {
            while (true) {
                canalConnector.connect();
                canalConnector.subscribe(".*\\..*"); // 订阅所有的数据库和表
                Message message = canalConnector.get(100); // 获取指定数量的数据

                if (message.getId() != -1) {
                    List<CanalEntry.Entry> entries = message.getEntries();
                    for (CanalEntry.Entry entry : entries) {
                        if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                            handleEntry(entry);
                        }
                    }
                }
                canalConnector.disconnect();
            }
        }).start();
    }

    private void handleEntry(CanalEntry.Entry entry) {
        // 处理数据变更
        CanalEntry.RowChange rowChange;
        try {
            rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (Exception e) {
            throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
        }

        CanalEntry.EventType eventType = rowChange.getEventType();
        String tableName = entry.getHeader().getTableName();

        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            if (eventType == CanalEntry.EventType.INSERT) {
                // 处理插入事件
                syncToRedis(rowData.getAfterColumnsList(), tableName);
            } else if (eventType == CanalEntry.EventType.UPDATE) {
                // 处理更新事件
                syncToRedis(rowData.getAfterColumnsList(), tableName);
            } else if (eventType == CanalEntry.EventType.DELETE) {
                // 处理删除事件
                deleteFromRedis(rowData.getBeforeColumnsList(), tableName);
            }
        }
    }

    private void syncToRedis(List<CanalEntry.Column> columns, String tableName) {
        // 这里根据业务逻辑同步到Redis中
        for (CanalEntry.Column column : columns) {
            redisTemplate.opsForHash().put(tableName, column.getName(), column.getValue());
        }
    }

    private void deleteFromRedis(List<CanalEntry.Column> columns, String tableName) {
        // 这里根据业务逻辑从Redis中删除数据
        for (CanalEntry.Column column : columns) {
            redisTemplate.opsForHash().delete(tableName, column.getName());
        }
    }
}
